package com.liuzy88;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;

public class P2P {
    // 控制台输入
    private static Scanner scanner = new Scanner(System.in);
    private static String cmd;
    private static String[] cmds;
    private static int len;
    // 运行模式 C/S
    private static String MODE;
    // 服务端
    private static String serverIp;
    private static int serverPort = 80;
    private static ConcurrentHashMap<String, Socket> serverClients = new ConcurrentHashMap<>();
    private static ConcurrentHashMap<String, PrintWriter> serverPws = new ConcurrentHashMap<>();
    // 客户端
    private static String localIp;
    private static Integer localPort;
    private static String me;
    private static Socket mySocket;
    private static PrintWriter myPw;
    private static BufferedReader myBr;
    private static ConcurrentHashMap<Integer, String> myClients = new ConcurrentHashMap<>();
    // p2p
    private static Socket p2pSocket;
    private static PrintWriter p2pPw;
    private static BufferedReader p2pBr;

    public static void main(String[] args) {
        System.out.println("******** P2P Test for JAVA ********");
        while (true) {
            String line = scanner.nextLine().trim();
            cmd = line.replaceAll("\\s+|\t+|\n| +", " ");
            cmds = cmd.split(" ");
            len = cmds.length;
            if ("S".equals(MODE)) {
                // 当前是服务端模式
                if ("view".equalsIgnoreCase(cmds[0])) {
                    if (serverClients.isEmpty()) {
                        System.out.println("Not any client connected.");
                    } else {
                        String str = toStr(serverClients);
                        System.out.print(str);
                    }
                } else {
                    System.out.println("Invalid command.");
                }
            } else if ("C".equals(MODE)) {
                // 当前是客户端模式
                if ("show".equalsIgnoreCase(cmd)) {
                    System.out.println("My local is " + localIp + ":" + localPort + ", I am " + me + " on server.");
                } else if ("view".equalsIgnoreCase(cmd)) {
                    if (myClients.isEmpty()) {
                        System.out.println("Is empty. Please pull before.");
                    } else {
                        for (int i = 0; i < myClients.size(); i++) {
                            String c = myClients.get(i);
                            if (c.equalsIgnoreCase(me)) {
                                System.out.println("[x] " + c + "  <- is own.");
                            } else {
                                System.out.println("[" + i + "] " + c);
                            }
                        }
                    }
                } else if ("p2p".equalsIgnoreCase(cmds[0])) {
                    int idx = toInt(cmds[1], -1);
                    String she;
                    if (idx < 0 || idx >= myClients.size() || (she = myClients.get(idx)) == null || me.equalsIgnoreCase(she)) {
                        System.out.println("Wrong target, choose another one.");
                    } else {
                        myPw.write("p2p>>" + me + "$" + she + "\n");
                        myPw.flush();
                    }
                } else if ("quit".equalsIgnoreCase(cmd) && p2pPw == null) {
                    myPw.write(cmd + "\n");
                    myPw.flush();
                    close(myBr, myPw, mySocket, p2pBr, p2pPw, p2pSocket);
                    System.exit(0);
                } else {
                    if (p2pPw != null) {
                        p2pPw.write(cmd + "\n");
                        p2pPw.flush();
                    } else {
                        myPw.write(cmd + "\n");
                        myPw.flush();
                    }
                }
            } else if (cmd.equalsIgnoreCase("listen") || cmd.startsWith("listen ")) {
                // 切换到服务端模式
                if (len == 2) {
                    serverPort = toInt(cmds[1], 80);
                }
                newThread("server-thread", () -> {
                    try {
                        ServerSocket serverSocket = new ServerSocket();
                        serverSocket.setReuseAddress(true);
                        serverSocket.bind(new InetSocketAddress(serverPort));
                        while (true) {
                            Socket socket = serverSocket.accept();
                            final String clientIP = socket.getInetAddress().getHostAddress();
                            final int clientPORT = socket.getPort();
                            final String client = clientIP + ":" + clientPORT;
                            serverClients.put(client, socket);
                            PrintWriter pw = new PrintWriter(socket.getOutputStream());
                            serverPws.put(client, pw);
                            newThread("client-thread_" + client, () -> {
                                try {
                                    System.out.println("Client " + client + " connected.");
                                    BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                                    pw.write("Connect success.\n");
                                    pw.write("me>>" + client + "\n");
                                    pw.flush();
                                    newThread("client-reader-thread_" + client, () -> {
                                        try {
                                            String msg;
                                            while (br != null && (msg = br.readLine().trim()) != null) {
                                                if ("quit".equalsIgnoreCase(msg)) {
                                                    System.out.println("The client " + client + " be quit.");
                                                    break;
                                                } else if ("pull".equalsIgnoreCase(msg)) {
                                                    String str = String.join("$", serverClients.keySet());
                                                    pw.write("pull>>" + str + "\n");
                                                    pw.flush();
                                                } else if (msg.startsWith("p2p>>")) {
                                                    // 有客户端请求p2p连接
                                                    msg = msg.substring("p2p>>".length());
                                                    String A = msg.split("\\$")[0];
                                                    String B = msg.split("\\$")[1];
                                                    System.out.println(A + " want p2p " + B);
                                                    PrintWriter Apw = serverPws.get(A);
                                                    PrintWriter Bpw = serverPws.get(B);
                                                    if (Apw != null && Bpw != null) {
                                                        System.out.println("Send \"p2p>>" + B + "\" to " + A);
                                                        Apw.write("p2p>>" + B + "\n");
                                                        Apw.flush();
                                                        System.out.println("Send \"p2p>>" + A + "\" to " + B);
                                                        Bpw.write("p2p>>" + A + "\n");
                                                        Bpw.flush();
                                                    } else {
                                                        System.out.println("Server can't help A and B.");
                                                    }
                                                } else {
                                                    System.out.println("Client " + client + " msg: " + msg);
                                                    pw.write("Send success.\n");
                                                    pw.flush();
                                                }
                                            }
                                        } catch (IOException e) {
                                            System.out.println("The client " + client + " exit. because " + e.getMessage());
                                        }
                                        serverClients.remove(client);
                                        serverPws.remove(client);
                                        close(br, pw, socket);
                                    });
                                } catch (Exception e) {
                                    e.printStackTrace();
                                    serverClients.remove(client);
                                    serverPws.remove(client);
                                    close(pw, socket);
                                }
                            });
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
                MODE = "S";
                System.out.println("The server is now ready on port " + serverPort + ".");
            } else if (cmd.startsWith("conn ") && len == 2) {
                // 切换到客户端模式
                if (!cmds[1].contains(":")) {
                    cmds[1] += ":80";
                }
                serverIp = cmds[1].split(":")[0];
                serverPort = Integer.parseInt(cmds[1].split(":")[1]);
                newThread("client-thread", () -> {
                    try {
                        mySocket = new Socket();
                        mySocket.setReuseAddress(true);
                        mySocket.connect(new InetSocketAddress(serverIp, serverPort));
                        localIp = mySocket.getLocalAddress().getHostAddress();
                        localPort = mySocket.getLocalPort();
                        myPw = new PrintWriter(mySocket.getOutputStream());
                        myBr = new BufferedReader(new InputStreamReader(mySocket.getInputStream()));
                        newThread("client-reader-thread", () -> {
                            try {
                                String msg;
                                while (!mySocket.isClosed() && myBr != null && (msg = myBr.readLine().trim()) != null) {
                                    if (msg.startsWith("me>>")) {
                                        // 服务端告诉我自己是谁
                                        me = msg.substring("me>>".length());
                                    } else if (msg.startsWith("pull>>")) {
                                        // 收到服务端的用户列表
                                        String[] rows = msg.substring("pull>>".length()).split("\\$");
                                        System.out.println("Received " + rows.length + " records.");
                                        myClients.clear();
                                        for (String row : rows) {
                                            myClients.put(myClients.size(), row);
                                        }
                                    } else if (msg.startsWith("p2p>>")) {
                                        // 服务器通知我去p2p她
                                        String she = msg.substring("p2p>>".length());
                                        System.out.println("P2p " + she + " connecting...");
                                        String sheIp = she.split(":")[0];
                                        int shePort = Integer.parseInt(she.split(":")[1]);
                                        newThread("p2p-thread_" + she, () -> {
                                            try {
                                                p2pSocket = new Socket();
                                                p2pSocket.setReuseAddress(true);
                                                p2pSocket.bind(new InetSocketAddress(localIp, localPort));
                                                p2pSocket.connect(new InetSocketAddress(sheIp, shePort));
                                                System.out.println("P2p connect success.");
                                                p2pPw = new PrintWriter(p2pSocket.getOutputStream());
                                                p2pBr = new BufferedReader(new InputStreamReader(p2pSocket.getInputStream()));
                                                newThread("p2p-reader-thread", () -> {
                                                    try {
                                                        String p2pMsg;
                                                        while (!p2pSocket.isClosed() && p2pBr != null && (p2pMsg = p2pBr.readLine().trim()) != null) {
                                                            if ("quit".equalsIgnoreCase(p2pMsg)) {
                                                                System.out.println("P2p connect quit.");
                                                                p2pPw.write("quit\n");
                                                                p2pPw.flush();
                                                                break;
                                                            } else if ("P2p Send success.".equals(p2pMsg)) {
                                                                System.out.println(p2pMsg);
                                                            } else {
                                                                System.out.println("P2p msg: " + p2pMsg);
                                                                p2pPw.write("P2p Send success.\n");
                                                                p2pPw.flush();
                                                            }
                                                        }
                                                    } catch (Exception e) {
                                                        e.printStackTrace();
                                                        close(p2pBr);
                                                    }
                                                    close(p2pBr, p2pPw, p2pSocket);
                                                    p2pPw = null;
                                                });
                                            } catch (Exception e) {
                                                System.out.println("p2p failed. " + e.getMessage());
                                                e.printStackTrace();
                                            }
                                        });
                                    } else {
                                        System.out.println(msg);
                                    }
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                                close(myBr);
                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                        close(myPw, mySocket);
                    }
                });
                MODE = "C";
            } else {
                System.out.println("listen <port>  or  conn <ip>:<port>");
            }
        }
    }

    public static void newThread(String name, Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setName(name);
        thread.start();
    }

    public static int toInt(String str, int defaultVal) {
        try {
            return Integer.parseInt(str.contains(":") ? str.split(":")[1].trim() : str.trim());
        } catch (Exception e) {
            return defaultVal;
        }
    }

    public static String toStr(ConcurrentHashMap<String, Socket> clients) {
        StringBuilder sb = new StringBuilder();
        int i = 0;
        for (String key : clients.keySet()) {
            sb.append("[" + ++i + "] " + key + "\n");
        }
        return sb.toString();
    }

    public static void close(Closeable... streams) {
        if (streams != null) {
            for (Closeable stream : streams) {
                try {
                    stream.close();
                } catch (Exception e) {
                    // pass
                }
            }
        }
    }
}

